package main

// 导入包
import (
	"log"
	"time"

	"github.com/streadway/amqp"
)

// 错误处理
func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}

func main() {
	// 连接RabbitMQ
	conn, err := amqp.Dial("amqp://guest:123456@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	// 创建信道
	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	// 声明要操作的队列
	q, err := ch.QueueDeclare(
		"hello", // 队列名需要跟发送消息的队列名保持一致
		true,    // durable
		false,   // delete when unused
		false,   // exclusive
		false,   // no-wait
		nil,     // arguments
	)
	failOnError(err, "Failed to declare a queue")
	// 创建消息消费者
	msgs, err := ch.Consume(
		q.Name,           // 队列名
		"hello_consumer", // 消费者名字，不填，则自动生成一个唯一ID
		false,            // 是否自动提交消息，即自动告诉rabbitmq消息已经处理成功。
		false,            // exclusive
		false,            // no-local
		false,            // no-wait
		nil,              // args
	)
	failOnError(err, "Failed to register a consumer")

	// 循环拉取队列中的消息
	for d := range msgs {
		// 打印消息内容
		retry := d.Headers["retry_count"].(int32)
		log.Printf("重试:%v次, 消息: %s", retry, d.Body)

		time.Sleep(1 * time.Second)
		// 确认消费
		// d.Ack(false)
		rejectToQueus(ch, &q, &d)
	}
}

// 重新放回队列，并更新重试次数
func rejectToQueus(ch *amqp.Channel, q *amqp.Queue, d *amqp.Delivery) {
	// 重新试3次不成功，丢弃
	if d.Headers["retry_count"].(int32) >= 3 {
		d.Reject(false)
		return
	}
	// 将消息发回队尾进行重试
	d.Headers["retry_count"] = d.Headers["retry_count"].(int32) + 1
	err := ch.Publish(
		"",     // exchange（交换机名字），这里忽略
		q.Name, // 路由参数，这里使用队列名字作为路由参数
		false,  // mandatory 强制
		false,  // immediate 实时
		amqp.Publishing{
			Headers:      d.Headers,
			DeliveryMode: d.DeliveryMode,
			ContentType:  d.ContentType,
			Body:         d.Body, // 消息内容
		})
	failOnError(err, "")
	d.Ack(false)
}
